tests/test_deploy.py (1)

19-47: Test coverage is good; consider adding formatExtension assertion in the last test.

The tests comprehensively cover multiple scenarios. However, the final test case (lines 44-46) validates only compression but doesn't verify that formatExtension is None as expected.

🔎 Suggested test completeness improvement

     # Test with compression only (no format extension)
     result = parse_distribution_str("http://example.com/data|.gz")
     assert result["url"] == "http://example.com/data"
     assert result["compression"] == "gz"
+    assert result["formatExtension"] is None
+    assert result["variants"] == {}

databusclient/cli.py (1)

15-57: Fix indentation and consider extracting the compression list.

1. Line 50: The print statement appears to have inconsistent indentation (5 spaces instead of 4). Verify this matches the project's style.

2. Line 37: The hardcoded compression extensions list is a pragmatic heuristic, but extracting it as a module-level constant would improve maintainability:

🔎 Suggested improvements

+# Common compression file extensions
+COMPRESSION_EXTENSIONS = {'.gz', '.zip', '.br', '.tar', '.zst'}
+
 def parse_distribution_str(dist_str: str):
     """
     Parses a distribution string with format:
     URL|key=value|...|.extension
     
     Returns a dictionary suitable for the deploy API.
     """
     parts = dist_str.split('|')
     url = parts[0].strip()
     
     variants = {}
     format_ext = None
     compression = None
     
     # Iterate over the modifiers (everything after the URL)
     for part in parts[1:]:
         part = part.strip()
         
         # Case 1: Extension (starts with .)
         if part.startswith('.'):
             # purely heuristic: if it looks like compression (gz, zip, br), treat as compression
             # otherwise treat as format extension
-            if part.lower() in ['.gz', '.zip', '.br', '.tar', '.zst']:
+            if part.lower() in COMPRESSION_EXTENSIONS:
                 compression = part.lstrip('.') # remove leading dot for API compatibility if needed
             else:
                 format_ext = part.lstrip('.')
         
         # Case 2: Content Variant (key=value)
         elif '=' in part:
             key, value = part.split('=', 1)
             variants[key.strip()] = value.strip()
             
         # Case 3: Standalone tag (treat as boolean variant or ignore? 
         # For now, we assume it's a value for a default key or warn)
         else:
-             print(f"WARNING: Unrecognized modifier '{part}' in distribution. Expected '.ext' or 'key=val'.")
+            print(f"WARNING: Unrecognized modifier '{part}' in distribution. Expected '.ext' or 'key=val'.")

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

databusclient/cli.py (1)

1-12: Past review comment about unused re import appears resolved.

The previous review mentioned an unused re import at line 4, but the current code shows from typing import List at that line. The issue appears to have been addressed.

databusclient/cli.py (2)

14-14: Add return type annotation.

The function should include a return type annotation for better type safety and documentation.

🔎 Proposed fix

+from typing import List, Dict, Any
+
-def parse_distribution_str(dist_str: str):
+def parse_distribution_str(dist_str: str) -> Dict[str, Any]:

---

49-49: Use click.echo() instead of print() for consistency.

For CLI applications using Click, prefer click.echo() over print() for better cross-platform compatibility and consistency with the rest of the CLI output.

🔎 Proposed fix

-             print(f"WARNING: Unrecognized modifier '{part}' in distribution. Expected '.ext' or 'key=val'.")
+             click.echo(f"WARNING: Unrecognized modifier '{part}' in distribution. Expected '.ext' or 'key=val'.")

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

tests/test_parse_distribution.py (4)

92-121: Consider adding a test for case-insensitive compression detection.

The implementation uses .lower() to match compression extensions (as seen in the relevant code snippet), but there's no test verifying that .GZ, .Zip, etc. are correctly detected. This ensures the case-insensitive behavior is maintained.

🔎 Suggested test addition

Add this test to the compression detection section:

def test_compression_case_insensitive(self):
    """Test that compression detection is case-insensitive."""
    result = parse_distribution_str("http://example.com/file|.GZ")
    assert result["compression"] == "gz"
    
    result = parse_distribution_str("http://example.com/file|.Zip")
    assert result["compression"] == "zip"

---

21-30: Consider adding a test for URLs containing pipe characters.

The current parsing logic splits on | to separate the URL from modifiers. If a URL contains a pipe character (though rare, it's valid in URLs), the parsing would break. Consider adding a test to document this limitation or suggest URL encoding.

🔎 Suggested test addition

Add this test to the URL extraction section:

def test_url_with_pipe_character(self):
    """Test handling of URLs containing pipe characters (edge case)."""
    # URLs can technically contain pipe characters
    # This test documents current behavior (would break parsing)
    # Consider URL encoding pipes as %7C in practice
    url_with_pipe = "http://example.com/data?filter=a%7Cb"  # %7C is URL-encoded pipe
    result = parse_distribution_str(f"{url_with_pipe}|lang=en")
    assert result["url"] == url_with_pipe

---

63-121: Consider using parametrized tests to reduce duplication.

The format extension tests (lines 63-87) and compression detection tests (lines 92-121) follow a repetitive pattern. Using pytest.mark.parametrize could make these more maintainable and concise.

🔎 Example parametrized test refactor

Replace the individual format extension tests with:

@pytest.mark.parametrize("extension,expected", [
    ("json", "json"),
    ("ttl", "ttl"),
    ("csv", "csv"),
    ("xml", "xml"),
])
def test_format_extension_detection(self, extension, expected):
    """Test format extension detection for various types."""
    result = parse_distribution_str(f"http://example.com/file|.{extension}")
    assert result["formatExtension"] == expected

Similarly for compression tests:

@pytest.mark.parametrize("compression,expected", [
    ("gz", "gz"),
    ("zip", "zip"),
    ("br", "br"),
    ("tar", "tar"),
    ("zst", "zst"),
])
def test_compression_detection(self, compression, expected):
    """Test compression detection for various types."""
    result = parse_distribution_str(f"http://example.com/file|.{compression}")
    assert result["compression"] == expected

---

126-146: Consider adding a test for multiple extensions of the same type.

The current implementation would overwrite if multiple format extensions (e.g., |.json|.xml) or multiple compression types (e.g., |.gz|.zip) are specified. A test documenting this behavior would help prevent confusion.

🔎 Suggested test addition

Add to the edge cases section:

def test_multiple_format_extensions_last_wins(self):
    """Test that when multiple format extensions are specified, the last one wins."""
    result = parse_distribution_str("http://example.com/file|.json|.xml")
    # Last extension specified should be used
    assert result["formatExtension"] == "xml"

def test_multiple_compression_types_last_wins(self):
    """Test that when multiple compression types are specified, the last one wins."""
    result = parse_distribution_str("http://example.com/file|.gz|.zip")
    # Last compression specified should be used
    assert result["compression"] == "zip"

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro